package com.xian.scala.stream

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}


object MyStreamWC2 {
  def main(args: Array[String]): Unit = {
    //传人参数  --host localhost --port 2323
    val parameterTool: ParameterTool = ParameterTool.fromArgs(args)

    val host: String = parameterTool.get("host")
    val port: Int = parameterTool.getInt("port")


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置并行度
    env.setParallelism(2)

    val lines: DataStream[String] = env.socketTextStream(host, port)

    val words: DataStream[String] = lines.flatMap(_.split(" "))

    val word_one: DataStream[(String, Int)] = words.map((_, 1))

    val word_n: KeyedStream[(String, Int), Tuple] = word_one.keyBy(0)

    val word_sum: DataStream[(String, Int)] = word_n.sum(1)

    //设置并行度
    word_sum.print().setParallelism(1)


    env.execute()

  }

}
